package com.cmsr.hdpf.etl.watermark

import com.cmsr.hdpf.model.RsmDataStream
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.watermark.Watermark

class PeriodicAssigner extends AssignerWithPeriodicWatermarks[RsmDataStream] {

  val bound =  5000 // 毫秒！ 窗口延迟关闭时间 -> 窗口延迟关闭失效？
  var maxTs = Long.MinValue + 2 * bound

  override def getCurrentWatermark: Watermark = {
//    println(maxTs, (new Watermark(maxTs - bound)).toString) // long.minvalue - bound 溢出导致一开始水印异常，值为 9223372036854770808！
    new Watermark(maxTs - bound) // 延迟触发
  }

  override def extractTimestamp(r: RsmDataStream, previousTS: Long): Long = {
//    println(Long.MinValue, maxTs, r.timestamp, this.getCurrentWatermark.toString, this.getCurrentWatermark.getTimestamp)
    maxTs = maxTs.max(r.timestamp)
    r.timestamp // 返回值
  }
}
